package org.springblade.protocol.mqtt;


import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * 初始化MQTT客户端
 * @author 李家民
 */
public class MqttInitialized {

	private static Map<String, MqttClient> clientMap = new HashMap<>(10);

	/**
	 * 服务连接
	 * @param address
	 * @param userName
	 * @param passwd
	 * @param app
	 */
	public static void connectServer(String address, String userName, String passwd, String app) {
		try {
			MqttConnectOptions connOpts = null;
			connOpts = new MqttConnectOptions();
			connOpts.setUserName(userName);
			connOpts.setPassword(passwd.toCharArray());
			connOpts.setAutomaticReconnect(true);
			connOpts.setCleanSession(true);
			//  持久化
			MemoryPersistence persistence = new MemoryPersistence();
			int nextInt = new Random().nextInt(50);
			MqttClient client = new MqttClient(address, app + nextInt, persistence);
			// 设置回调
			client.setCallback(new MqttCallbackExtended() {
				@Override
				public void connectionLost(Throwable throwable) {
					System.out.println("connection lost：" + throwable.getMessage());
				}

				@Override
				public void messageArrived(String topic, MqttMessage message) throws Exception {
					int qos = message.getQos();
					String data = new String(message.getPayload());
					System.out.println("Received message: \n  topic：" + topic
						+ "\n  Qos：" + qos
						+ "\n  payload：" + data);
				}

				@Override
				public void deliveryComplete(IMqttDeliveryToken token) {
					try {
						System.out.println("消息传递成功---deliveryComplete:\n" + token.getMessageId() + "    " + new String(token.getMessage().getPayload()));
					} catch (MqttException e) {
						e.printStackTrace();
					}
				}

				@Override
				public void connectComplete(boolean b, String serverURI) {
					System.out.println("MQTT连接成功:" + serverURI);
				}
			});
			// 建立连接
			client.connect(connOpts);

			// 放在map 另外做一个监听
			clientMap.put(address + ":" + userName, client);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	/**
	 * 消息订阅
	 * @param address
	 * @param userName
	 * @param topic
	 * @param qos
	 */
	public static void subscribe(String address, String userName, String topic, Integer qos) {
		try {
			MqttClient mqttClient = clientMap.get(address + ":" + address);
			mqttClient.subscribe(topic, qos);
		} catch (MqttException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 消息发布
	 * @param address
	 * @param userName
	 * @param topic
	 * @param message
	 * @param qos
	 */
	public static void publish(String address, String userName, String topic, String message, Integer qos) {
		MqttMessage mqttMessage = new MqttMessage(message.getBytes());
		mqttMessage.setQos(qos);
		try {
			MqttClient mqttClient = clientMap.get(address + ":" + address);
			mqttClient.publish(topic, mqttMessage);
		} catch (MqttException e) {
			throw new RuntimeException(e);
		}
	}

}
